from confluent_kafka import Consumer, KafkaError
import json
from config.config import settings
from dify_api import dify_api
from queue_push.queue_push import QueuePush
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
import threading
import asyncio
import aiohttp
from collections import OrderedDict


class KafkaConsumer:
    def __init__(self):
        self.consumer = Consumer({
            'bootstrap.servers': settings.queue_push_sever,
            'group.id': settings.group_id,
            'enable.auto.commit': False  # 关闭自动提交
        })
        self.topic = settings.getdata_topic_name
        self.consumer.subscribe([settings.getdata_topic_name])
        self.api_query = dify_api.DifyAPIQuery()
        self.executor = ThreadPoolExecutor(max_workers=settings.max_workers)  # 可以根据需要调整线程数
        self.user_question_record = OrderedDict()  # 使用OrderedDict来存储用户问题记录
        self.clean_up_interval = 120  # 清理任务间隔时间，以秒为单位，这里设置为120秒
        self.record_expiry_time = 60  # 用户记录有效期，以秒为单位，这里设置为60秒

        # 启动定时清理任务，设为守护线程使其在主线程结束时自动结束
        self.clean_up_thread = threading.Thread(target=self.clean_up_expired_records, daemon=True)
        self.clean_up_thread.start()  # 启动清理线程

    # 新增异步方法，用于发送异步API请求
    async def fetch_response(self, user_question, user_id):
        async with aiohttp.ClientSession() as session:
            return await self.api_query.send_query(session, user_question, user_id)

    def process_message(self, message_data, msg):
        try:
            event_time_str = message_data.get('event_time')
            if event_time_str:
                event_time = datetime.strptime(event_time_str, '%Y-%m-%dT%H:%M:%S')
                current_time = datetime.now()
                time_diff = abs((current_time - event_time).total_seconds())

                if time_diff > settings.max_time_interval:
                    print(f"Message event time is too old: {event_time_str} current_time: {current_time}")
                    return  # 超过10秒的不处理

            user_question = message_data.get('user_question', 'No question provided')
            user_id = message_data.get('user_id')

            # 检查是否在规定时间内收到相同的问题
            if user_id in self.user_question_record:
                last_question, last_time = self.user_question_record[user_id]
                if last_question == user_question and (current_time - last_time).total_seconds() < 10:
                    # 如果在规定时间内收到相同问题，不处理并返回
                    print(f"Duplicate question received from user {user_id} within 10 seconds.")
                    return
                else:
                    # 否则更新用户问题和时间
                    self.user_question_record.move_to_end(user_id)  # 将更新的记录移到末尾
                    self.user_question_record[user_id] = (user_question, current_time)
            else:
                # 新用户或新的问题，存储用户问题和时间
                self.user_question_record[user_id] = (user_question, current_time)

            # 创建一个新的事件循环并在该循环中运行异步任务
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            response_content = new_loop.run_until_complete(self.fetch_response(user_question, user_id))
            new_loop.close()

            print(response_content["answer"])

            # 增加回复内容字段
            if settings.reply_indicate_user:
                message_data['intelligent_reply'] = f"@{message_data.get('user_name') }"+response_content["answer"]
            else:
                message_data['intelligent_reply'] = response_content["answer"]
            print(message_data)

            # 丢入消息队列
            push = QueuePush(message_data)
            push.push()

            # 手动提交偏移量
            self.consumer.commit(message=msg)
        except Exception as e:
            print(f"Error processing message: {e}")

    def start_consuming(self):
        try:
            print(f"Starting to consume messages from {self.topic}")
            futures = []

            while True:
                msg = self.consumer.poll(1.0)  # 从Kafka中拉取消息，超时时间1秒

                if msg is None:
                    continue  # 没有消息则继续轮询
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        # End of partition event
                        continue  # 分区结束事件，继续轮询
                    else:
                        print(msg.error())  # 打印错误信息
                        break  # 处理其它错误

                # 处理消息
                print('Received message: {}'.format(json.loads(msg.value())))
                message_data = json.loads(msg.value())
                future = self.executor.submit(self.process_message, message_data, msg)
                futures.append(future)

                # 清理已完成的任务
                futures = [future for future in futures if not future.done()]

                # 检查并提交已完成的任务的偏移量
                for future in as_completed(futures):
                    try:
                        future.result()  # 检查并获取任务结果
                    except Exception as e:
                        print(f"Error in future: {e}")  # 处理任务中的异常

        except KeyboardInterrupt:
            print("Stopped by user.")  # 用户手动中断
        finally:
            self.close()  # 清理资源

    def close(self):
        print("Closing consumer.")
        self.consumer.close()  # 关闭消费者
        self.executor.shutdown(wait=True)  # 关闭线程池并等待任务完成

    def clean_up_expired_records(self):
        while True:
            current_time = datetime.now()
            expired_keys = []
            for user_id, (_, record_time) in self.user_question_record.items():
                # 检查记录是否过期
                if (current_time - record_time).total_seconds() > self.record_expiry_time:
                    expired_keys.append(user_id)
                else:
                    break  # 因为OrderedDict是有序的，所以可以提前退出循环
            for key in expired_keys:
                del self.user_question_record[key]  # 删除过期记录
            threading.Event().wait(self.clean_up_interval)  # 等待指定时间后再次清理
